package com.bd08.flink.demo

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.assigners.{GlobalWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

import scala.util.Try

object Xg  extends App {


  case class Record(date:String,i:Double)

  val env = StreamExecutionEnvironment.getExecutionEnvironment

  env.readTextFile("D:\\yahu\\flink-demo2\\src\\main\\resources\\xg\\nhc-all.csv")
    .filter(line => !line.contains("日期,新增确诊"))
    .map(line => {
      val arr = line.split(",")
      arr(4) = (arr(8).toInt - Try(arr(7).toInt).getOrElse(0) - Try(arr(6).toInt).getOrElse(0)).toString
//      var rs = ""
//      // String.join
//      arr.foreach(e => rs = rs +","+e)
//      rs.substring(1)
      // 拆箱
      val Array(date,xin,_,_,xian,_,_,_,lie,_*)=arr
      Record(date.substring(0,date.lastIndexOf("/")),(lie.toInt-Try(xin.toInt).getOrElse(0))/lie.toDouble)
    })
    .keyBy(_.date)
    //.windowAll(GlobalWindows.create())
    //.window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    //.sum("xin")
    .maxBy("i")
    .print()
  env.execute()


}
